package com.github.yoojia.flow;

import com.github.yoojia.flow.util.Wrap;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.github.yoojia.flow.util.Invoker.catchInvoke;

/**
 * 负责消息路由。
 *
 * @author 2017 Yoojia Chen (yoojiachen@gmail.com)
 */
final class Dispatcher extends InitialContext implements Lifecycle {

    private static final Logger LOGGER = LoggerFactory.getLogger(Dispatcher.class);

    private static final int EVENT_QUEUE_CAPACITY = 64;

    private final List<EventInterceptor> mInterceptors;
    private final List<EventPipeline> mPipelines;
    private final List<EventHandler> mHandlers;
    private final List<EventDelivery> mDeliveries;

    private final ExecutorService mEventHandlerThreads = Executors.newCachedThreadPool();

    private final Disruptor<EventW> mDisruptor = new Disruptor<>(
            EventW.factory(),
            EVENT_QUEUE_CAPACITY,
            mEventHandlerThreads,
            ProducerType.SINGLE,
            new BlockingWaitStrategy());

    private final List<OnEventReceivedListener> mOnEventReceivedListeners = new CopyOnWriteArrayList<>();

    ////

    @SuppressWarnings("unchecked")
    Dispatcher(List<EventInterceptor> interceptors,
               List<EventPipeline> pipelines,
               List<EventHandler> handlers,
               List<EventDelivery> deliveries) {
        mInterceptors = interceptors;
        mPipelines = pipelines;
        mHandlers = handlers;
        mDeliveries = deliveries;

        // 设置事件处理器
        mDisruptor.handleEventsWith((EventW evt, long sequence, boolean e) -> {
            // 处理完成消息后，由事件Listener做监听回调处理
            final Event processed = processEvent(evt.getEvent());
            mOnEventReceivedListeners
                    .forEach(listener ->
                            listener.onEvent(processed));
        });
    }

    @Override
    public void onStart() {
        mDisruptor.start();
    }

    @Override
    public void onStop() {
        try {
            mDisruptor.shutdown();
        } finally {
            if (!mEventHandlerThreads.isShutdown()) {
                mEventHandlerThreads.shutdown();
            }
        }
    }

    public void addOnEventReceivedListener(OnEventReceivedListener listener) {
        mOnEventReceivedListeners.add(listener);
    }

    public void removeOnEventReceivedListener(OnEventReceivedListener listener) {
        mOnEventReceivedListeners.remove(listener);
    }

    /**
     * 将事件压入队列。
     * 派发器区分发送给本地设备的事件和发给远程设备的事件，使用两个队列来缓存，以加速本地设备事件的处理速度。
     *
     * @param event Event
     */
    void enqueueEvent(Event event) {
        final RingBuffer<EventW> buffer = mDisruptor.getRingBuffer();
        final long evtId = buffer.next();
        // 填充事件负载
        try {
            buffer.get(evtId)
                    .setEvent(event);
        } finally {
            buffer.publish(evtId);
        }
    }

    /**
     * 派发处理消息事件。在线程池中执行。
     *
     * 消息被派发时进入Dispatcher，都会经过几个步骤处理：
     * 1. Interceptor 拦截组，任何一个消息被拦截后，消息将被中止后续处理。
     * 2. Pipeline 消息转换处理。
     * 3. Handler 消息动作执行。
     * 4. Delivery 派发至目标设备。
     *
     * @param raw Event
     */
    private Event processEvent(Event raw) {
        // 事件拦截：
        // 任何一个拦截器都可以对事件拦截。
        if (!mInterceptors.isEmpty() &&
                mInterceptors.parallelStream()
                        .anyMatch(interceptor -> {
                            try {
                                return interceptor.onInterceptEvent(getContext(), raw);
                            } catch (Throwable e) {
                                LOGGER.error("Error when EventInterceptor processing", e);
                                // 拦截器内部发生错误时，拦截事件，后续不再处理。
                                return true;
                            }
                        })) {
            // 拦截后，不处理其它流程
            return Event.simple(raw.sequence, Context.TOPIC_SYS_FILTERED);
        }

        final Wrap<Event> wrapper = new Wrap<>(raw);

        // 事件转换：
        // 事件转换器根据自身判断，将事件转换成新事件，后续事件流程将使用新的事件继续处理。
        mPipelines.forEach(pipeline -> {
            final Event newEvent = catchInvoke(() -> {
                final Event ret = pipeline.onProcessEvent(getContext(), wrapper.get());
                if (ret == null) {
                    throw new IllegalArgumentException("EventPipeline MUST NOT return null Event object");
                }
                // ！！ 确保返回Event的序列号ID是请求的ID：
                if (ret.sequence == raw.sequence) {
                    return ret;
                } else {
                    return ret.newBuilder()
                            .setSequence(raw.sequence)
                            .build();
                }
            }, "Error when EventPipeline processing");
            if (newEvent != null) {
                wrapper.set(newEvent);
            }
        });

        final Event processedEvent = wrapper.get();

        // 事件处理：
        // 在事件处理过程中，事件处理器不会对当前事件的流程做任何改变，但可以通过发送其它事件来作出响应。
        mHandlers.parallelStream()
                .forEach(handler ->
                        catchInvoke(() -> {
                            handler.onHandleEvent(getContext(), processedEvent);
                            return null;
                        }, "Error when EventHandler processing"));

        // 事件派发：
        final boolean send = mDeliveries.parallelStream()
                .anyMatch((delivery -> {
                    final Event ret = catchInvoke(() ->
                                    delivery.onDeliverEvent(getContext(), processedEvent),
                            "Error when EventDelivery processing");
                    // 发生内部错误时，ret为null
                    if (ret != null) {
                        wrapper.set(ret);
                    }
                    // 任意一个非Skip的响应事件，表明事件已被派发。
                    return ret != null &&
                            !Context.TOPIC_SYS_SKIPPED.equals(ret.topic);
                }));

        if (send) {
            // 在返回处理结果时，无论返回的Event是否设置sequence ID，
            // 在这里都会被设置为请求的sequence ID：
            return wrapper.get();
        } else {
            return Event.simple(raw.sequence, Context.TOPIC_SYS_SKIPPED);
        }
    }

    public interface OnEventReceivedListener {
        void onEvent(Event event);
    }
}
